package com.ruyuan.engine.core;
import com.alibaba.fastjson.JSON;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.utils.Constants;
import com.ruyuan.engine.utils.FlinkKafkaUtils;
import com.ruyuan.engine.utils.HBaseUtils;
import com.ruyuan.engine.utils.RedisUtils;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.HTable;
import org.locationtech.spatial4j.distance.DistanceUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

/**
 * 第一版需求：
 *      1. 完成是否异地登录判断
 * */
public class ProcessBaseRule {
    protected static final Logger LOG = LoggerFactory.getLogger(ProcessBaseRule.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //1.获取Kafka数据源
        DataStreamSource<String> source = environment.addSource(FlinkKafkaUtils.getKafkaEventSource());
        SingleOutputStreamOperator<ClientLog> source2 = source.map(m -> JSON.parseObject(m, ClientLog.class));

        source2.keyBy(s -> s.getUserNo())
                .process(new KeyedProcessFunction<String, ClientLog, String>() {
                    Jedis jedis;

                    /**
                     * 从hbase读取用户画像数据
                     * */
                    HTable table;

                    //基于状态存储用户上一次登录的数据
                    ListState<ClientLog> privousLoginData;

                    //缓存用户登录历史数据
                    ListState<ClientLog> historyLoginData;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        jedis = RedisUtils.getJedis();
                        table = HBaseUtils.initHbaseClient(Constants.HBASE_TABLE);
                        privousLoginData = getRuntimeContext().getListState(new ListStateDescriptor<ClientLog>("privousLoginData",ClientLog.class));
                        historyLoginData = getRuntimeContext().getListState(new ListStateDescriptor<ClientLog>("historyLoginData",ClientLog.class));
                    }

                    //对每条数据进行处理
                    @Override
                    public void processElement(ClientLog clientLog, Context ctx, Collector<String> out) {
                        String eventType = clientLog.getEventType();
                        if (eventType.equals(Constants.RULE_TYPE_LOGIN)) {
                            /**
                             * 1.5分钟登录次数限制
                             * 2.异地登录
                             * 3.不在常用地区登录
                             * */

                          //缓存登录数据
//                         cacheLoginData(clientLog,historyLoginData);

                         //异地登录
//                         distanceProcess(clientLog,privousLoginData);

                         //5分钟登录次数限制
//                         minTimesProcess(clientLog,historyLoginData);

                         /**
                          *
                          * 是否在常用地区登录
                          * 1. 用户常用地 => 来自于画像系统
                          * 2. 犯罪高发地区：缓存到redis
                          * */
                         locationRange(clientLog,jedis);
                        }else if(eventType.equals(Constants.RULE_TYPE_OPER)) {
                            /**
                             * 修改手机号
                             * */
                            changePhoneProcess(clientLog);
                        }
                    }
                }).print();
        environment.execute("process1");
    }

    /**
     * 修改手机号
     * */
    private static void changePhoneProcess(ClientLog clientLog) {
        String score = "0";
        String reason = "";
        //常用登录城市
        String historyCitys = HBaseUtils.getCell(Constants.HBASE_TABLE, clientLog.getUserNo(), "region", "historyloc");
        //登录设备
        String imei = HBaseUtils.getCell(Constants.HBASE_TABLE, clientLog.getUserNo(), "registration", "imei");
        String city = clientLog.getAddr().substring(clientLog.getAddr().indexOf("省")+1,clientLog.getAddr().indexOf("市"));
        if (clientLog.getEventId().equals(Constants.OPER_CHANGE_PHONE)) {
            /**
             * 是否在常用设备，是否异地
             * */
            //不在常用地区，并且异地
            if (!(clientLog.getImei().equals(imei) && historyCitys.contains(city))) {
                score = "1";
                reason = "当前登录城市："+city+" 不在历史登录地: "+historyCitys+" 不在常用登录设备："+imei;
            }else if(clientLog.getImei().equals(imei) && !historyCitys.contains(city)) {
                score = "3";
                reason = "当前登录城市："+city+" 不在历史登录地: "+historyCitys+" 在常用登录设备："+imei;
            }else if(!clientLog.getImei().equals(imei) && historyCitys.contains(city)) {
                score = "2";
                reason = "当前登录城市："+city+" 在历史登录地: "+historyCitys+" 不在常用登录设备："+imei;
            }else  {
                score = "5";
                reason = "当前登录城市："+city+" 在历史登录地: "+historyCitys+" 在常用登录设备："+imei;
            }
        }
        LOG.info("score={} reason = {}",score,reason);
    }

    /**
     * 是否在常用地区登录
     * */
    private static void locationRange(ClientLog clientLog, Jedis jedis) {
        //获取用户历史登录城市 , 深圳|北京|上海|杭州
        String historyCitys = HBaseUtils.getCell(Constants.HBASE_TABLE, clientLog.getUserNo(), "region", "historyloc");
        if (historyCitys == null || historyCitys.isEmpty()) {
            historyCitys = "深圳|北京|上海|杭州";
        }

        //犯罪高发地区：在redis 北京|广州|杭州
        String dangerCitys = jedis.get("dangerCitys");

        //定义返回值
        String score = "0";
        String reason = "未知原因";
        //判断当前用户的登录地址
        if (clientLog.getAddr() !=null && !clientLog.getAddr().isEmpty()) {
            //正常登录
            String city = clientLog.getAddr().substring(clientLog.getAddr().indexOf("省") + 1, clientLog.getAddr().indexOf("市"));
            if (historyCitys.length() == 0) {
                //第一次登录
                score = "5";
                reason = "当前登录城市：" + city + " 是第一次登录";
            }else {
                /**
                 * 非第一次登录
                 *      是否在常用历史城市
                 *         不在
                 *             在犯罪高发地区 1
                 *             不在犯罪高发地区 2
                 *         在
                 *             在犯罪高发地区 3
                 *             不在犯罪高发地区 5
                 * */
                //判断是否在常用历史城市 , 在常用历史登录城市
                if (historyCitys.contains(city)) {
                    //在犯罪高发地区
                    if (dangerCitys.contains(city)) {
                        score = "3";
                        reason = "当前登录城市"+city+"  在常用历史登录城市"+ historyCitys+"  在犯罪高发地区："+dangerCitys;
                    }else {
                        //不在犯罪高发地区
                        score = "5";
                        reason = "当前登录城市"+city+"  在常用历史登录城市"+ historyCitys+"  不在犯罪高发地区："+dangerCitys;
                    }
                }else {
                    //不在常用历史登录城市
                    //在犯罪高发地区
                    if (dangerCitys.contains(city)) {
                        score = "1";
                        reason = "当前登录城市"+city+"  不在常用历史登录城市"+ historyCitys+"  在犯罪高发地区："+dangerCitys;
                    }else {
                        //不在犯罪高发地区
                        score = "2";
                        reason = "当前登录城市"+city+"  不在常用历史登录城市"+ historyCitys+"  不在犯罪高发地区："+dangerCitys;
                    }
                }
            }
        }else {
            //当前登录城市信息获取异常
            score = "0";
            reason = "当前登录城市异常";
        }
        LOG.info("score = {}  reason={}  ",score,reason);
    }

    /**
     * 5分钟登录次数限制
     * */
    private static void minTimesProcess(ClientLog clientLog, ListState<ClientLog> historyLoginData) {
        try {
            Iterable<ClientLog> clientLogs = historyLoginData.get();
            List<ClientLog> lists = IteratorUtils.toList(clientLogs.iterator());

            /**
             * 规则
             *      5：限制次数
             *      5：分数1
             *      2：分数2
             * */
            String ruleLogin = "5|5|2";
            String limitTimes = ruleLogin.split("\\|")[0];
            String score1 = ruleLogin.split("\\|")[1];
            String score2 = ruleLogin.split("\\|")[2];

            //判断登录次数+得分
            Integer loginCount = 1;
            if (lists != null && !lists.isEmpty()) {
                loginCount = lists.size();
            }

            String score = "0";
            String reason = "未知原因";
            if (loginCount >= Integer.parseInt(limitTimes)) {
                score = score2;
                reason = "5min内的登录次数为： "+loginCount + " 大于了规定的次数: "+limitTimes;
            }else {
                score = score1;
                reason = "5min内的登录次数为： "+loginCount + " 小于了规定的次数: "+limitTimes;
            }
            System.out.println("得分："+score+" 原因："+reason);
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 缓存登录数据
     * 只缓存5分钟以内的登录数据
     * */
    private static void cacheLoginData(ClientLog clientLog, ListState<ClientLog> historyLoginData) {
        try {
            //获取5分钟以内用户登录的数据
            Iterable<ClientLog> clientLogs = historyLoginData.get();
            //清空缓存数据
            historyLoginData.clear();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
            long currentTime = new Date().getTime();
            //5分钟之前
            long startTime = sdf.parse(sdf.format(currentTime - 5 * 60 * 1000)).getTime();
            List<ClientLog> lists = IteratorUtils.toList(clientLogs.iterator());
            if (lists != null && !lists.isEmpty()) {
                for(ClientLog log:lists) {
                    long logTime = sdf.parse(sdf.format(Long.parseLong(log.getDetailTime()))).getTime();
                    if (logTime > startTime && logTime < currentTime) {
                        historyLoginData.add(log);
                    }
                }
            }
            historyLoginData.add(clientLog);
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 异地登录
     * */
    public static void distanceProcess(ClientLog clientLog, ListState<ClientLog> privousLoginData) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        String score = "0";
        String reason = "第一次登录: "+clientLog.getAddr()+" 本次登录时间："+clientLog.getDetailTime();
        try {
            //判断上一次是否登录
            Iterator<ClientLog> iterator = privousLoginData.get().iterator();
            if (iterator.hasNext()) {
                System.out.println(" 不是第一次 ");
                ClientLog privousMessage = iterator.next();
                //不是第一次登录
                //静态规则
                String distanceRule = "500|1@400|2@300|3@200|4@100|5";
                String[] distanceRules = distanceRule.split("@");

                String oldTime = privousMessage.getDetailTime();
                String oldGps = privousMessage.getGps();
                String oldAddr = privousMessage.getAddr();

                //计算距离
                double distanceReal = DistanceUtils.distHaversineRAD(Double.parseDouble(clientLog.getGps().split("\\,")[1]), Double.parseDouble(clientLog.getGps().split("\\,")[0]), Double.parseDouble(oldGps.split("\\,")[1]), Double.parseDouble(oldGps.split("\\,")[1])) * DistanceUtils.EARTH_EQUATORIAL_RADIUS_KM;
                //时间差
//                long time = sdf.parse(clientLog.getDetailTime()).getTime() - sdf.parse(oldTime).getTime();
                long time = sdf.parse(sdf.format(Long.parseLong(clientLog.getDetailTime()))).getTime() - sdf.parse(sdf.format(Long.parseLong(oldTime))).getTime();
                double speed = distanceReal / (time / (1000 * 3600.0));
                //测试初始值
//                speed = 369d;
                //规则匹配
                for(String str:distanceRules) {
                    double speedLimit = Double.parseDouble(str.split("\\|")[0]);
                    String speedScore = str.split("\\|")[1];
                    if (speed <= speedLimit) {
                        score = speedScore;
                        reason = "=== 短时间内速度为："+speed+"  规则约定速度为："+speedLimit+" 当前登录地："+clientLog.getAddr()+" 上一次登录地："+privousMessage.getAddr();
                    }
                }
            }else {
                System.out.println(" 是第一次 ");
                //第一次登录
                score = "5";
                reason = "第一次登录: "+clientLog.getAddr()+" 登录时间为："+clientLog.getDetailTime();
            }
            privousLoginData.clear();
            privousLoginData.add(clientLog);
            System.out.println("score ="+score + " reason ="+reason);

        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}